package com.kyu.LogDiff;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Iterator;

public class Reduce extends Reducer<Text,Bean,Text,NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<Bean> values, Context context) throws IOException, InterruptedException {
        String result = key.toString().replace(" ",",");
        Iterator it = values.iterator();
        ArrayList<Bean> beanList= new ArrayList<>();
        while (it.hasNext()) beanList.add((Bean)it.next());     // 为了重复遍历，将values的元素填充到ArrayList中
        result += "," + beanList.get(0).getTemper_diffdew();
        Double temper_diff[] = new Double[4];
        Double press_diff[] = new Double[4];
//        System.out.print(key.toString() + " ");
//        for (Bean bean : beanList) System.out.print(bean.getOffset() + " ");
//        System.out.println();
        Integer vaild = beanList.stream().filter(x->x.getOffset() == 0).findFirst().orElse(new Bean()).getOffset();
        int size = beanList.size();
        if (vaild == null || size < 2) return;                              // 利用offset属性判断无效数据
        int i = 0;
        for (Bean bean : beanList) {
            temper_diff[i] = bean.getTemper();
            result += "," + temper_diff[i];
            press_diff[i] = bean.getPress();
            i++;
        }
        BigDecimal bd_temper = new BigDecimal(Double.toString(temper_diff[0]));
        BigDecimal bd_press = new BigDecimal(Double.toString(press_diff[0]));
        for (i = 1; i < 4; i++) {
            if (temper_diff[i] == null || press_diff[i] == null) continue;
            temper_diff[i] = new BigDecimal(Double.toString(temper_diff[i])).subtract(bd_temper).doubleValue();
            press_diff[i] = new BigDecimal(Double.toString(press_diff[i])).subtract(bd_press).doubleValue();
        }   // 输出格式：s_date,s_time,temper,temper_3h,temper_6h,temper_24h,press,press_3h,press_6h,press_24h
        for (int j = 0; j < 4; j++) result += "," + press_diff[j];
//        System.out.println(result);
        context.write(new Text(result),null);
    }
}
